package message_client

import (
	"context"
	"encoding/json"
	"github.com/satori/go.uuid"
	"github.com/segmentio/kafka-go"
	"strings"
)

type KafkaMessageClient struct {
	Producer *kafka.Writer
}

func NewKafkaMessageClient(kafkaURL []string, topic string) *KafkaMessageClient {
	return &KafkaMessageClient{
		Producer: kafka.NewWriter(kafka.WriterConfig{
			Brokers: kafkaURL,
			Topic:   topic,
		}),
	}
}

func NewKafkaMessageClientWithConfig(config kafka.WriterConfig) *KafkaMessageClient {
	return &KafkaMessageClient{
		Producer: kafka.NewWriter(config),
	}
}

func (messageClient *KafkaMessageClient) SendSms(businessName string, parameter *SmsRequestParameter) (result *SendResults, err error) {
	requestParameter := buildSendParameter(businessName, parameter.Delay, parameter.SendTime, MESSAGE_TYPE_SMS)
	requestParameter.Sms = parameter
	return messageClient.doRequest(requestParameter)
}

func (messageClient *KafkaMessageClient) SendAppNotification(businessName string, parameter *AppNotificationRequestParameter) (result *SendResults, err error) {
	requestParameter := buildSendParameter(businessName, parameter.Delay, parameter.SendTime, MESSAGE_TYPE_APP_NOTIFICATION)
	requestParameter.AppNotification = parameter
	return messageClient.doRequest(requestParameter)
}

func (messageClient *KafkaMessageClient) SendAppTransmission(businessName string, parameter *AppTransmissionRequestParameter) (result *SendResults, err error) {
	requestParameter := buildSendParameter(businessName, parameter.Delay, parameter.SendTime, MESSAGE_TYPE_APP_TRANSMISSION)
	requestParameter.AppTransmission = parameter
	return messageClient.doRequest(requestParameter)
}

func (messageClient *KafkaMessageClient) SendMail(businessName string, parameter *MailRequestParameter) (result *SendResults, err error) {
	requestParameter := buildSendParameter(businessName, parameter.Delay, parameter.SendTime, MESSAGE_TYPE_EMAIL)
	requestParameter.Email = parameter
	return messageClient.doRequest(requestParameter)
}

func (messageClient *KafkaMessageClient) SendInternalMessage(businessName string, parameter *InternalMessageRequestParameter) (result *SendResults, err error) {
	requestParameter := buildSendParameter(businessName, parameter.Delay, parameter.SendTime, MESSAGE_TYPE_INTERNAL_MESSAGE)
	requestParameter.InternalMessage = parameter
	return messageClient.doRequest(requestParameter)
}

func (messageClient *KafkaMessageClient) SendDingTalkMessage(businessName string, parameter *DingtalkMessageRequest) (result *SendResults, err error) {
	requestParameter := buildSendParameter(businessName, parameter.Delay, parameter.SendTime, MESSAGE_TYPE_DING_TALK_MESSAGE)
	requestParameter.DingtalkMessage = parameter
	return messageClient.doRequest(requestParameter)
}

func (messageClient *KafkaMessageClient) SendExtendPlatformMessage(
	businessName string, args map[string]interface{}, delay bool, sendTime int64) (result *SendResults, err error) {

	argsBytes, err := json.Marshal(args)
	parameter := &ExtendPlatformMessageRequest{Args: string(argsBytes)}

	requestParameter := buildSendParameter(businessName, delay, sendTime, MESSAGE_TYPE_EXTEND_PLATFORM_MESSAGE)
	requestParameter.ExtendPlatformMessage = parameter

	return messageClient.doRequest(requestParameter)
}

func (messageClient *KafkaMessageClient) Send(businessName string, messageTypes []string, requestParameter *SendRequestParameter) (result *SendResults, err error) {
	requestId := uuid.NewV4().String()
	requestId = strings.ReplaceAll(requestId, "-", "")
	requestParameter.RequestId = requestId
	requestParameter.BusinessName = businessName
	requestParameter.MessageTypes = messageTypes
	return messageClient.doRequest(requestParameter)
}

func (messageClient *KafkaMessageClient) doRequest(requestParameter *SendRequestParameter) (result *SendResults, err error) {

	if byteArr, err := json.Marshal(requestParameter); err == nil {
		if err := messageClient.Producer.WriteMessages(context.Background(), kafka.Message{Value: byteArr}); err == nil {
			return &SendResults{
				Code:      SUCCESS,
				RequestId: requestParameter.RequestId,
			}, err
		} else {
			return &SendResults{
				Message:   err.Error(),
				RequestId: requestParameter.RequestId,
			}, err
		}
	} else {
		return &SendResults{
			Message:   err.Error(),
			RequestId: requestParameter.RequestId,
		}, err
	}
}

/*
type KafkaConfig struct {
	Brokers           []string
	Topic             string
	MaxAttempts       int
	QueueCapacity     int
	BatchSize         int
	BatchBytes        int
	BatchTimeout      time.Duration
	ReadTimeout       time.Duration
	WriteTimeout      time.Duration
	RebalanceInterval time.Duration
	IdleConnTimeout   time.Duration
	RequiredAcks      int
	Async             bool
}

type Dialer struct {
	ClientID        string
	Timeout         time.Duration
	Deadline        time.Time
	LocalAddr       net.Addr
	DualStack       bool
	FallbackDelay   time.Duration
	KeepAlive       time.Duration
	TLS             *tls.Config
	SASLMechanism   sasl.Mechanism
	TransactionalID string
}
*/
